#include "dtu_data_push.h"
#include "dtu_log.h"
#include "dtu_memory.h"
#include "dtu_mqtt.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

/* 全局数据推送管理器 */
static push_manager_t g_push_manager = {0};

/* 内部函数声明 */
static void* push_thread_func(void* arg);
static dtu_error_t push_task_mqtt(const push_task_t* task);
static dtu_error_t push_task_http(const push_task_t* task);
static dtu_error_t push_task_tcp(const push_task_t* task);
static dtu_error_t push_task_udp(const push_task_t* task);
static dtu_error_t add_task_to_queue(push_task_t* task);
static push_task_t* get_next_task(void);
static void free_task(push_task_t* task);
static void free_task_queue(void);
static char* generate_task_id(void);

dtu_error_t dtu_data_push_init(const push_config_t* config)
{
    if (g_push_manager.initialized) {
        return DTU_OK;
    }

    if (!config) {
        return DTU_ERROR_INVALID_PARAM;
    }

    memset(&g_push_manager, 0, sizeof(push_manager_t));

    /* 复制配置 */
    g_push_manager.config = *config;

    /* 初始化互斥锁和条件变量 */
    if (pthread_mutex_init(&g_push_manager.mutex, NULL) != 0) {
        return DTU_ERROR;
    }

    if (pthread_cond_init(&g_push_manager.cond, NULL) != 0) {
        pthread_mutex_destroy(&g_push_manager.mutex);
        return DTU_ERROR;
    }

    /* 初始化统计信息 */
    g_push_manager.stats.start_time = time(NULL);

    g_push_manager.initialized = true;

    DTU_LOG_INFO("Data push manager initialized with protocol: %s", 
                 dtu_push_protocol_string(config->protocol));
    return DTU_OK;
}

void dtu_data_push_cleanup(void)
{
    if (!g_push_manager.initialized) {
        return;
    }

    /* 停止推送服务 */
    dtu_data_push_stop();

    pthread_mutex_lock(&g_push_manager.mutex);

    /* 释放任务队列 */
    free_task_queue();

    g_push_manager.initialized = false;

    pthread_mutex_unlock(&g_push_manager.mutex);
    pthread_mutex_destroy(&g_push_manager.mutex);
    pthread_cond_destroy(&g_push_manager.cond);

    DTU_LOG_INFO("Data push manager cleanup completed");
}

dtu_error_t dtu_data_push_start(void)
{
    if (!g_push_manager.initialized) {
        return DTU_ERROR;
    }

    pthread_mutex_lock(&g_push_manager.mutex);

    if (g_push_manager.running) {
        pthread_mutex_unlock(&g_push_manager.mutex);
        return DTU_OK;
    }

    /* 根据协议初始化连接 */
    if (g_push_manager.config.protocol == PUSH_PROTOCOL_MQTT) {
        mqtt_config_t mqtt_config = {
            .port = g_push_manager.config.server_port,
            .keepalive = g_push_manager.config.keepalive,
            .clean_session = g_push_manager.config.clean_session
        };
        DTU_STRNCPY(mqtt_config.broker, g_push_manager.config.server_host, sizeof(mqtt_config.broker));
        DTU_STRNCPY(mqtt_config.client_id, g_push_manager.config.client_id, sizeof(mqtt_config.client_id));
        DTU_STRNCPY(mqtt_config.username, g_push_manager.config.username, sizeof(mqtt_config.username));
        DTU_STRNCPY(mqtt_config.password, g_push_manager.config.password, sizeof(mqtt_config.password));
        DTU_STRNCPY(mqtt_config.topic_prefix, g_push_manager.config.topic, sizeof(mqtt_config.topic_prefix));

        if (dtu_mqtt_init(&mqtt_config) != DTU_OK) {
            pthread_mutex_unlock(&g_push_manager.mutex);
            DTU_LOG_ERROR("Failed to initialize MQTT client");
            return DTU_ERROR;
        }

        if (dtu_mqtt_connect() != DTU_OK) {
            pthread_mutex_unlock(&g_push_manager.mutex);
            DTU_LOG_ERROR("Failed to connect to MQTT broker");
            return DTU_ERROR_NETWORK;
        }

        g_push_manager.connected = true;
    }

    g_push_manager.running = true;

    /* 创建推送线程 */
    if (pthread_create(&g_push_manager.push_thread, NULL, push_thread_func, NULL) != 0) {
        g_push_manager.running = false;
        g_push_manager.connected = false;
        if (g_push_manager.config.protocol == PUSH_PROTOCOL_MQTT) {
            dtu_mqtt_disconnect();
            dtu_mqtt_cleanup();
        }
        pthread_mutex_unlock(&g_push_manager.mutex);
        DTU_LOG_ERROR("Failed to create push thread");
        return DTU_ERROR;
    }

    pthread_mutex_unlock(&g_push_manager.mutex);

    DTU_LOG_INFO("Data push service started");
    return DTU_OK;
}

void dtu_data_push_stop(void)
{
    if (!g_push_manager.initialized || !g_push_manager.running) {
        return;
    }

    pthread_mutex_lock(&g_push_manager.mutex);
    g_push_manager.running = false;
    pthread_cond_signal(&g_push_manager.cond);
    pthread_mutex_unlock(&g_push_manager.mutex);

    /* 等待推送线程结束 */
    pthread_join(g_push_manager.push_thread, NULL);

    /* 断开连接 */
    if (g_push_manager.config.protocol == PUSH_PROTOCOL_MQTT) {
        dtu_mqtt_disconnect();
        dtu_mqtt_cleanup();
    }

    g_push_manager.connected = false;

    DTU_LOG_INFO("Data push service stopped");
}

dtu_error_t dtu_data_push_data(const collect_result_t* result, bool urgent)
{
    if (!g_push_manager.initialized || !result) {
        return DTU_ERROR_INVALID_PARAM;
    }

    /* 创建推送任务 */
    push_task_t* task = DTU_MALLOC(sizeof(push_task_t));
    if (!task) {
        return DTU_ERROR_MEMORY;
    }

    memset(task, 0, sizeof(push_task_t));

    /* 生成任务ID */
    char* task_id = generate_task_id();
    if (task_id) {
        DTU_STRNCPY(task->task_id, task_id, sizeof(task->task_id));
        DTU_FREE(task_id);
    }

    /* 复制采集数据 */
    task->data = DTU_MALLOC(sizeof(collect_result_t));
    if (!task->data) {
        DTU_FREE(task);
        return DTU_ERROR_MEMORY;
    }

    *task->data = *result;

    /* 复制数据值数组 */
    if (result->value_count > 0 && result->values) {
        task->data->values = DTU_MALLOC(result->value_count * sizeof(data_value_t));
        if (!task->data->values) {
            DTU_FREE(task->data);
            DTU_FREE(task);
            return DTU_ERROR_MEMORY;
        }
        memcpy(task->data->values, result->values, result->value_count * sizeof(data_value_t));
    }

    task->config = g_push_manager.config;
    task->create_time = time(NULL);
    task->urgent = urgent;

    /* 添加到任务队列 */
    dtu_error_t ret = add_task_to_queue(task);
    if (ret != DTU_OK) {
        dtu_data_collect_free_result(task->data);
        DTU_FREE(task->data);
        DTU_FREE(task);
        return ret;
    }

    DTU_LOG_DEBUG("Data push task created: %s (device=%s, urgent=%s)", 
                  task->task_id, result->device_name, urgent ? "yes" : "no");
    return DTU_OK;
}

dtu_error_t dtu_data_push_get_stats(push_stats_t* stats)
{
    if (!g_push_manager.initialized || !stats) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_push_manager.mutex);
    *stats = g_push_manager.stats;
    pthread_mutex_unlock(&g_push_manager.mutex);

    return DTU_OK;
}

bool dtu_data_push_is_connected(void)
{
    if (!g_push_manager.initialized) {
        return false;
    }

    pthread_mutex_lock(&g_push_manager.mutex);
    bool connected = g_push_manager.connected;
    pthread_mutex_unlock(&g_push_manager.mutex);

    return connected;
}

dtu_error_t dtu_data_push_format_json(const collect_result_t* result, char* json_str, size_t size)
{
    if (!result || !json_str || size == 0) {
        return DTU_ERROR_INVALID_PARAM;
    }

    /* 简化的MQTT消息格式，只包含timestamp(ms)和data */
    int len = snprintf(json_str, size,
                      "{"
                      "\"timestamp\":%ld,",
                      result->collect_time * 1000);  /* 转换为毫秒 */

    if (len >= (int)size) {
        return DTU_ERROR_MEMORY;
    }

    /* 如果有原始16进制数据，直接使用原始数据 */
    if (result->has_raw_data) {
        len += snprintf(json_str + len, size - len,
                       "\"data\":\"%s\"",
                       result->raw_hex_data);
    }

    /* 结束JSON对象 */
    if (len < (int)size - 10) {
        strcat(json_str, "}");
    }

    return DTU_OK;
}

const char* dtu_push_protocol_string(push_protocol_t protocol)
{
    switch (protocol) {
        case PUSH_PROTOCOL_MQTT: return "MQTT";
        case PUSH_PROTOCOL_HTTP: return "HTTP";
        case PUSH_PROTOCOL_TCP: return "TCP";
        case PUSH_PROTOCOL_UDP: return "UDP";
        default: return "Unknown";
    }
}

const char* dtu_data_format_string(data_format_t format)
{
    switch (format) {
        case DATA_FORMAT_JSON: return "JSON";
        case DATA_FORMAT_XML: return "XML";
        case DATA_FORMAT_CSV: return "CSV";
        case DATA_FORMAT_BINARY: return "Binary";
        default: return "Unknown";
    }
}

/* 内部函数实现 */
static void* push_thread_func(void* arg)
{
    (void)arg;

    DTU_LOG_INFO("Data push thread started");

    while (g_push_manager.running) {
        pthread_mutex_lock(&g_push_manager.mutex);

        /* 等待任务或超时 */
        if (!g_push_manager.task_queue) {
            struct timespec timeout;
            clock_gettime(CLOCK_REALTIME, &timeout);
            timeout.tv_sec += 1; /* 1秒超时 */
            pthread_cond_timedwait(&g_push_manager.cond, &g_push_manager.mutex, &timeout);
        }

        /* 获取下一个任务 */
        push_task_t* task = get_next_task();
        pthread_mutex_unlock(&g_push_manager.mutex);

        if (!task) {
            continue;
        }

        /* 执行推送任务 */
        dtu_error_t ret = DTU_ERROR;
        switch (task->config.protocol) {
            case PUSH_PROTOCOL_MQTT:
                ret = push_task_mqtt(task);
                break;
            case PUSH_PROTOCOL_HTTP:
                ret = push_task_http(task);
                break;
            case PUSH_PROTOCOL_TCP:
                ret = push_task_tcp(task);
                break;
            case PUSH_PROTOCOL_UDP:
                ret = push_task_udp(task);
                break;
            default:
                DTU_LOG_ERROR("Unsupported push protocol: %d", task->config.protocol);
                ret = DTU_ERROR_NOT_SUPPORTED;
                break;
        }

        pthread_mutex_lock(&g_push_manager.mutex);

        if (ret == DTU_OK) {
            /* 推送成功 */
            g_push_manager.stats.successful_tasks++;
            g_push_manager.stats.last_push_time = time(NULL);
            DTU_LOG_DEBUG("Task %s pushed successfully", task->task_id);
        } else {
            /* 推送失败，检查是否需要重试 */
            task->retry_count++;
            if (task->retry_count < task->config.retry_count) {
                task->retry_time = time(NULL) + task->config.retry_interval;
                /* 重新加入队列 */
                task->next = g_push_manager.task_queue;
                g_push_manager.task_queue = task;
                g_push_manager.stats.retry_tasks++;
                DTU_LOG_WARN("Task %s failed, will retry (%d/%d)",
                            task->task_id, task->retry_count, task->config.retry_count);
                pthread_mutex_unlock(&g_push_manager.mutex);
                continue;
            } else {
                /* 重试次数用完，放弃任务 */
                g_push_manager.stats.failed_tasks++;
                DTU_LOG_ERROR("Task %s failed after %d retries", task->task_id, task->retry_count);
            }
        }

        g_push_manager.stats.total_tasks++;
        pthread_mutex_unlock(&g_push_manager.mutex);

        /* 释放任务 */
        free_task(task);
    }

    DTU_LOG_INFO("Data push thread stopped");
    return NULL;
}

static dtu_error_t push_task_mqtt(const push_task_t* task)
{
    if (!task || !task->data) {
        return DTU_ERROR_INVALID_PARAM;
    }

    /* 格式化数据 */
    char json_data[4096];
    dtu_error_t ret = dtu_data_push_format_json(task->data, json_data, sizeof(json_data));
    if (ret != DTU_OK) {
        DTU_LOG_ERROR("Failed to format data to JSON");
        return ret;
    }

    /* 直接使用配置的Topic，不要自己拼接 */
    const char* topic = task->config.topic;

    /* 发布消息 */
    ret = dtu_mqtt_publish(topic, json_data, strlen(json_data), MQTT_QOS_1, false);
    if (ret != DTU_OK) {
        DTU_LOG_ERROR("Failed to publish MQTT message");
        return ret;
    }

    g_push_manager.stats.total_bytes += strlen(json_data);
    return DTU_OK;
}

static dtu_error_t push_task_http(const push_task_t* task)
{
    /* TODO: 实现HTTP推送 */
    (void)task;
    DTU_LOG_WARN("HTTP push not implemented yet");
    return DTU_ERROR_NOT_SUPPORTED;
}

static dtu_error_t push_task_tcp(const push_task_t* task)
{
    /* TODO: 实现TCP推送 */
    (void)task;
    DTU_LOG_WARN("TCP push not implemented yet");
    return DTU_ERROR_NOT_SUPPORTED;
}

static dtu_error_t push_task_udp(const push_task_t* task)
{
    /* TODO: 实现UDP推送 */
    (void)task;
    DTU_LOG_WARN("UDP push not implemented yet");
    return DTU_ERROR_NOT_SUPPORTED;
}

static dtu_error_t add_task_to_queue(push_task_t* task)
{
    if (!task) {
        return DTU_ERROR_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_push_manager.mutex);

    /* 添加到队列头部 */
    task->next = g_push_manager.task_queue;
    g_push_manager.task_queue = task;
    g_push_manager.stats.pending_tasks++;

    /* 通知推送线程 */
    pthread_cond_signal(&g_push_manager.cond);

    pthread_mutex_unlock(&g_push_manager.mutex);
    return DTU_OK;
}

static push_task_t* get_next_task(void)
{
    push_task_t* task = NULL;
    push_task_t* prev = NULL;
    push_task_t* current = g_push_manager.task_queue;
    time_t now = time(NULL);

    /* 查找可执行的任务（优先级：紧急任务 > 重试时间到期的任务 > 普通任务） */
    while (current) {
        if (current->urgent || current->retry_time <= now) {
            /* 从队列中移除 */
            if (prev) {
                prev->next = current->next;
            } else {
                g_push_manager.task_queue = current->next;
            }
            task = current;
            g_push_manager.stats.pending_tasks--;
            break;
        }
        prev = current;
        current = current->next;
    }

    return task;
}

static void free_task(push_task_t* task)
{
    if (!task) {
        return;
    }

    if (task->data) {
        dtu_data_collect_free_result(task->data);
        DTU_FREE(task->data);
    }

    DTU_FREE(task);
}

static void free_task_queue(void)
{
    push_task_t* task = g_push_manager.task_queue;
    while (task) {
        push_task_t* next = task->next;
        free_task(task);
        task = next;
    }
    g_push_manager.task_queue = NULL;
    g_push_manager.stats.pending_tasks = 0;
}

static char* generate_task_id(void)
{
    char* task_id = DTU_MALLOC(64);
    if (!task_id) {
        return NULL;
    }

    snprintf(task_id, 64, "task_%ld_%d", time(NULL), rand());
    return task_id;
}
